package com.xiguanlezz.cn.common.remote.client.grpc;

import com.google.common.util.concurrent.ListenableFuture;
import com.xiguanlezz.cn.api.common.Constants;
import com.xiguanlezz.cn.api.grpc.auto.BiRequestStreamGrpc;
import com.xiguanlezz.cn.api.grpc.auto.Payload;
import com.xiguanlezz.cn.api.grpc.auto.RequestGrpc;
import com.xiguanlezz.cn.api.remote.request.ConnectionSetupRequest;
import com.xiguanlezz.cn.api.remote.request.Request;
import com.xiguanlezz.cn.api.remote.request.ServerCheckRequest;
import com.xiguanlezz.cn.api.remote.request.SetupAckRequest;
import com.xiguanlezz.cn.api.remote.response.Response;
import com.xiguanlezz.cn.api.remote.response.ServerCheckResponse;
import com.xiguanlezz.cn.api.remote.response.SetupAckResponse;
import com.xiguanlezz.cn.common.remote.ConnectionType;
import com.xiguanlezz.cn.common.remote.client.Connection;
import com.xiguanlezz.cn.common.remote.client.RpcClient;
import com.xiguanlezz.cn.common.remote.client.ServerListFactory;
import com.xiguanlezz.cn.common.remote.client.ServerRequestHandler;
import com.xiguanlezz.cn.common.utils.ThreadFactoryBuilder;
import io.grpc.CompressorRegistry;
import io.grpc.DecompressorRegistry;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.*;

/**
 * @Author：习惯了种种
 * @CreateDate：2024/11/19 20:21
 * @Description：GRPC客户端通信对象
 */
@Slf4j
public class GrpcClient extends RpcClient {
    private GrpcClientConfig clientConfig /* GRPC客户端配置类 */;
    private ThreadPoolExecutor grpcExecutor; /* GRPC线程池 */

    // 等待服务端建立连接的同步器
    private final RecAbilityContext recAbilityContext = new RecAbilityContext(null);
    // 服务端连接建立后，服务端还需要将客户端支持的能力点发给客户端。这个对象就是该请求的处理器
    // 因为SetupRequestHandler是个内部类，所以放在这里
    private SetupRequestHandler setupRequestHandler;


    public GrpcClient(String name) {
        this(new DefaultGrpcClientConfig(name));
    }

    public GrpcClient(String name, Map<String, String> labels) {
        this(new DefaultGrpcClientConfig(name, labels));
    }

    public GrpcClient(GrpcClientConfig clientConfig) {
        super(clientConfig);
        this.clientConfig = clientConfig;
        initSetupHandler();
    }

    public GrpcClient(GrpcClientConfig clientConfig, ServerListFactory serverListFactory) {
        super(clientConfig, serverListFactory);
        this.clientConfig = clientConfig;
        initSetupHandler();
    }

    public String getName() {
        return clientConfig.name();
    }

    private void initSetupHandler() {
        // register to handler setup request
        setupRequestHandler = new SetupRequestHandler(this.recAbilityContext);
    }

    private ThreadPoolExecutor createGrpcExecutor(String serverIp) {
        // TODO(xiguanlezz)：放开注释？
        // serverIp = serverIp.replaceAll("%", "-");
        // 在这里创建了线程池对象，定义了线程池的核心线程数，最大线程数等等
        ThreadPoolExecutor grpcExecutor = new ThreadPoolExecutor(clientConfig.threadPoolMaxSize(),
                clientConfig.threadPoolMaxSize(), clientConfig.threadPoolKeepAlive(), TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(clientConfig.threadPoolQueueSize()),
                new ThreadFactoryBuilder().daemon(true).namePrefix("nacos-grpc-client-executor-" + serverIp)
                        .build());
        grpcExecutor.allowCoreThreadTimeOut(true);
        return grpcExecutor;
    }

    private ManagedChannel createNewManagedChannel(String serverIp, int serverPort) {
        return ManagedChannelBuilder.forAddress(serverIp, serverPort)
                .executor(grpcExecutor) /* 显示设置请求的执行线程池，如果不显示指定，那就是使用GRPC内置的线程池 */
                .compressorRegistry(CompressorRegistry.getDefaultInstance())
                .decompressorRegistry(DecompressorRegistry.getDefaultInstance())
                .maxInboundMessageSize(clientConfig.maxInboundMessageSize())
                .keepAliveTime(clientConfig.channelKeepAlive(), TimeUnit.MILLISECONDS)
                .keepAliveTimeout(clientConfig.channelKeepAliveTimeout(), TimeUnit.MILLISECONDS)
                .usePlaintext().build();
    }

    private RequestGrpc.RequestFutureStub createNewFutureChannelStub(ManagedChannel managedChannel) {
        return RequestGrpc.newFutureStub(managedChannel);
    }

    private Response serverCheck(String ip, int port, RequestGrpc.RequestFutureStub requestFutureStub) throws ExecutionException, InterruptedException, TimeoutException {
        try {
            ServerCheckRequest serverCheckRequest = new ServerCheckRequest();
            Payload grpcReq = GrpcUtils.convert(serverCheckRequest);
            ListenableFuture<Payload> listenableFuture = requestFutureStub.request(grpcReq);
            Payload response = listenableFuture.get(clientConfig.serverCheckTimeOut(), TimeUnit.MILLISECONDS);
            return (Response) GrpcUtils.parse(response);
        } catch (Exception e) {
            log.info("Server check fail, please check server：{}, port：{} is available, error = {}", ip, port, e);
            return null;
        }
    }

    private void shutdownChannel(ManagedChannel managedChannel) {
        if (managedChannel != null && !managedChannel.isShutdown()) {
            managedChannel.shutdownNow();
        }
    }

    @Override
    public Connection connectToServer(ServerInfo serverInfo) {
        try {
            // 1. 创建GRPC请求处理线程池
            if (Objects.isNull(grpcExecutor)) {
                // 线程池用在异步请求发送给服务端，处理回调用
                this.grpcExecutor = createGrpcExecutor(serverInfo.getServerIp());
            }

            // 2. 获取Nacos服务端用于服务注册的端口
            // 注意此时serverInfo.getServerPort()只是暴露CRUD的端口，用于服务注册的端口和这个不一样
            int port = serverInfo.getServerPort() + rpcPortOffset();

            // 3. 创建连接到远端的代理对象
            // 注意：GRPC连接建立是在使用代理对象给服务端发请求时
            ManagedChannel managedChannel = createNewManagedChannel(serverInfo.getServerIp(), port);
            // requestFutureStub可以用来监听异步回调线程,
            // 同时requestFutureStub这玩意儿就是个代理对象，代理普通的request方法，让RPC调用像本地方法调用一样
            RequestGrpc.RequestFutureStub requestFutureStub = createNewFutureChannelStub(managedChannel);

            // 4. 发送一个普通请求，只是探测服务端是否存活
            Response response = serverCheck(serverInfo.getServerIp(), port, requestFutureStub);
            if (!response.isSuccess()) {
                // GG...
                shutdownChannel(managedChannel);
                log.error("Server fail to response to [{}]！", GrpcClient.this.getName());
                return null;
            }

            // Nacos服务端流式响应处理器
            StreamObserver<Payload> responseStreamObserver = new StreamObserver<Payload>() {
                @Override
                public void onNext(Payload payload) {
                    try {

                        Object parseBody = GrpcUtils.parse(payload);
                        Request request = (Request) parseBody;
                        if (Objects.isNull(request)) {
                            return;
                        }

                        try {
                            // 这里为什么单独拆分？
                            // 因为SetupAckRequest这个请求对应的处理器是内部类，无法统一处理
                            if (request instanceof SetupAckRequest) {
                                // there is no connection ready this time
                                setupRequestHandler.requestReply(request, null);
                                return;
                            }

                            // 使用通用的服务端请求处理器处理
                            Response response = handleServerRequest(request);
                            if (Objects.nonNull(response)) {
                                response.setRequestId(request.getRequestId());
                                sendResponse(response);
                            }

                        } catch (Exception e) {
                        }

                    } catch (Exception e) {
                    }
                }

                @Override
                public void onError(Throwable throwable) {

                }

                @Override
                public void onCompleted() {

                }
            };

            // 5. 建立双向流式处理的通道
            BiRequestStreamGrpc.BiRequestStreamStub grpcNormalServiceStub = BiRequestStreamGrpc.newStub(requestFutureStub.getChannel());
            // biStreamObserver这玩意儿是客户端和服务端双向流式发消息的入口
            StreamObserver<Payload> biStreamObserver = grpcNormalServiceStub.requestBiStream(responseStreamObserver);

            ServerCheckResponse serverCheckResponse = (ServerCheckResponse) response;

            // 6. 封装发送请求处理类
            GrpcConnection grpcConnection = new GrpcConnection(serverInfo);
            grpcConnection.setConnectionId(serverCheckResponse.getConnectionId());
            grpcConnection.setChannel(managedChannel);
            grpcConnection.setGrpcNormalServiceStub(requestFutureStub);
            grpcConnection.setBiStreamObserver(biStreamObserver);
            grpcConnection.setExecutor(grpcExecutor);
            ConnectionSetupRequest setupRequest = new ConnectionSetupRequest();
            setupRequest.setLabels(super.getLabels());

            if (serverCheckResponse.isSupportAbilityNegotiation()) {
                // 初始化同步器对象
                this.recAbilityContext.reset(grpcConnection);
                // 连接支持的能力表都由服务端返回
                grpcConnection.setAbilityTable(null);
            }

            // 7. 用双向通道给服务端发消息，用于服务端注册客户端连接
            // 服务端接收到ConnectionSetupRequest请求后，就会在服务端的GrpcBiStreamRequestAcceptor类中发送一个SetupAckRequest请求给客户端，作为ConnectionSetupRequest请求的响应
            grpcConnection.sendRequest(setupRequest);
            if (recAbilityContext.needToSync) {
                // 阻塞
                recAbilityContext.await(clientConfig.capabilityNegotiationTimeout(), TimeUnit.MILLISECONDS);
            }
            return grpcConnection;
        } catch (ExecutionException | InterruptedException | TimeoutException e) {
            log.error("[{}] fail to connect to server！error = {}", GrpcClient.this.getName(), e);
            return null;
        }
    }

    private void sendResponse(Response response) {
        try {
            ((GrpcConnection) this.currentConnection).sendResponse(response);
        } catch (Exception e) {
            log.error("[{}] error to send ack response, ackId->{}", this.currentConnection.getConnectionId(),
                    response.getRequestId());
        }
    }

    /**
     * 客户端响应SetupAckRequest的处理器
     */
    class SetupRequestHandler implements ServerRequestHandler {

        private final RecAbilityContext abilityContext;

        public SetupRequestHandler(RecAbilityContext abilityContext) {
            this.abilityContext = abilityContext;
        }

        @Override
        public Response requestReply(Request request, Connection connection) {
            // if finish setup
            if (request instanceof SetupAckRequest) {
                SetupAckRequest setupAckRequest = (SetupAckRequest) request;
                // remove and count down
                recAbilityContext.release(
                        Optional.ofNullable(setupAckRequest.getAbilityTable()).orElse(new HashMap<>(0)));
                return new SetupAckResponse();
            }
            return null;
        }
    }

    /**
     * 简单的同步器，用于客户端初次启动给服务端发送ConnectionSetupRequest请求，让服务端建立连接时使用
     */
    static class RecAbilityContext {
        private volatile Connection connection;
        private volatile CountDownLatch blocker;
        private volatile boolean needToSync = false;

        public RecAbilityContext(Connection connection) {
            this.connection = connection;
            this.blocker = new CountDownLatch(1);
        }

        public boolean isNeedToSync() {
            return this.needToSync;
        }

        public void reset(Connection connection) {
            this.connection = connection;
            this.blocker = new CountDownLatch(1);
            this.needToSync = true;
        }

        public void release(Map<String, Boolean> abilities) {
            if (this.connection != null) {
                this.connection.setAbilityTable(abilities);
                // avoid repeat setting
                this.connection = null;
            }
            if (this.blocker != null) {
                blocker.countDown();
            }
            this.needToSync = false;
        }

        public void await(long timeout, TimeUnit unit) throws InterruptedException {
            if (this.blocker != null) {
                this.blocker.await(timeout, unit);
            }
            this.needToSync = false;
        }

        public boolean check(Connection connection) {
            if (!connection.isAbilitiesSet()) {
                log.error("Client don't receive server abilities table even empty table but server supports ability negotiation."
                                + " You can check if it is need to adjust the timeout of ability negotiation by property: {}"
                                + " if always fail to connect.",
                        GrpcConstants.GRPC_CHANNEL_CAPABILITY_NEGOTIATION_TIMEOUT);
                connection.setAbandon(true);
                connection.close();
                return false;
            }
            return true;
        }
    }

    @Override
    public ConnectionType getConnectionType() {
        return ConnectionType.GRPC;
    }

    @Override
    public int rpcPortOffset() {
        return Integer.parseInt(
                System.getProperty(
                        GrpcConstants.NACOS_SERVER_GRPC_PORT_OFFSET_KEY,
                        String.valueOf(Constants.SDK_GRPC_PORT_DEFAULT_OFFSET)
                )
        );
    }

}
